package com.hzh.flink.core

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object Demo9WindowProcess {
  def main(args: Array[String]): Unit = {
    /**
     * 创建一个环境
     */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    /**
     * 读取socket的数据
     */
    val lines: DataStream[String] = env.socketTextStream("master", 8888)

    val keyByDS: KeyedStream[String, String] = lines.flatMap(_.split(","))
      .keyBy(word => word)

    //统计5秒单词的数量
    val windowDS: WindowedStream[String, String, TimeWindow] = keyByDS
      .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))

    /**
     * 在窗口使用process函数，一次传入一个窗口数据
     *
     */

    val countDS: DataStream[(String, Long, Long, Int)] = windowDS.process(new ProcessWindowFunction[String, (String, Long, Long, Int), String, TimeWindow] {
      /**
       * process:一个窗口处理一次
       *
       * @param key      ：key
       * @param context  ：上下文对象，可以获取窗口的开始和结束时间
       * @param elements ：窗口内所有数据
       * @param out      ：用于将数据发送到下游
       */
      override def process(key: String, context: Context, elements: Iterable[String],
                           out: Collector[(String, Long, Long, Int)]): Unit = {

        //计算单词的数量
        val count: Int = elements.size
        //获取窗口的开始和结束时间
        val window: TimeWindow = context.window
        val startTime: Long = window.getStart
        val endTime: Long = window.getEnd

        //将统计的结果发送到下游

        out.collect((key, startTime, endTime, count))
      }
    })

    countDS.print()
    env.execute()

  }

}
